package lab.poc.erl;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

// Seems to be functional now (when did it not?), but i'm
// a bit concerned about its performance characteristics.
// Should maybe be redrafted in a future version.
final class ErlangStyleMailboxQueue {

	/*
	 * Linked list node class.
	 */
	private static final class Node {
		private Mailbox.MessageWrapper item;

		// One of:
		// - the successor Node
		// - null, meaning there is no successor (this is the last node)
		private Node next;

		private Node(final Mailbox.MessageWrapper value) {
			item = value;
		}
	}

	/*
	 * Number of elements in the queue available for reading. Not necessarily
	 * equal to the total number of elements in the queue, since Erlang keeps
	 * old unmatched messages in the mailbox and enables re-reading them only
	 * after a new message arrived that could be matched successfully.
	 */
	private final AtomicInteger numAvailable = new AtomicInteger();

	/* Head of linked list */
	private Node head;

	/* Tail of linked list */
	private Node last;

	private volatile boolean isLocked = false;

	private Pattern[] lastPatternSeen;

	/*
	 * Current read position in the queue. Elements to the left are older
	 * unmatched messages that won't get read again until this pointer is reset
	 * to head after a new message arrives that could be matched successfully.
	 */
	private Node curr;

	/* Lock held by take/etc */
	private final ReentrantLock takeLock = new ReentrantLock();

	/* Wait condition for waiting takes */
	private final Condition available = takeLock.newCondition();

	/* Lock held by put/etc */
	private final ReentrantLock putLock = new ReentrantLock();

	public ErlangStyleMailboxQueue() {
		init();
	}

	/**
	 * Atomically removes all of the elements from this queue.
	 * The queue will be empty after this call returns.
	 * @throws InterruptedException
	 */
	public void clear() throws InterruptedException {
		putLock.lockInterruptibly();
		// Note: takeLock not strictly needed in our single reader-thread scenario
		try {
			init();
		} finally {
			putLock.unlock();
		}
	}

	private void init() {
		last = curr = head = new Node(null);
		lastPatternSeen = null;
	}

	void shutdown() {
		boolean interrupted = false;
		while (!isLocked) {
			final ReentrantLock putLock = this.putLock;
			try {
				putLock.lockInterruptibly();
				try {
					isLocked = true;
					init(); // let GC do its work			
				} finally {
					putLock.unlock();
				}
			} catch (InterruptedException e) {
				interrupted = true;
			}
		}
		if (interrupted) {
			Thread.currentThread().interrupt();
		}
	}

	private void resetIfPatternsChanged(Pattern... predicates) {
		final Pattern[] lastPattern = lastPatternSeen;
		if (lastPattern == null || lastPattern.length != predicates.length) {
			lastPatternSeen = predicates;
			curr = head;
			return;
		}
		for (int i = 0; i < lastPattern.length; ++i) {
			// object equality is the only way to compare functions
			if (!lastPattern[i].equals(predicates[i])) {
				lastPatternSeen = predicates;
				curr = head;
				break;
			}
		}
	}

	/*
	 * Signals a waiting take. Called only from put (which does not otherwise
	 * lock takeLock).
	 */
	private void signalNewNodeAvailable() {
		final ReentrantLock takeLock = this.takeLock;
		takeLock.lock();
		try {
			available.signal();
		} finally {
			takeLock.unlock();
		}
	}

	/**
	 * Inserts the specified element at the tail of this queue.
	 * 
	 * @throws InterruptedException
	 * @throws IllegalArgumentException
	 */
	public void put(final Mailbox.MessageWrapper value) throws InterruptedException {
		if (value == null) {
			throw new IllegalArgumentException();
		}
		if (isLocked) {
			return;
		}
		// Note: convention in all put/take/etc methods is to preset local
		// counter variable holding a negative count to indicate failure unless
		// set.
		int c = -1;
		final ReentrantLock putLock = this.putLock;
		final AtomicInteger availableCount = this.numAvailable;
		putLock.lockInterruptibly();
		try {
			// create the new node and link it at the end of the queue
			last = last.next = new Node(value);
			c = availableCount.getAndIncrement();
		} finally {
			putLock.unlock();
		}
		if (c >= 0) {
			signalNewNodeAvailable();
		}
	}

	/*
	 * Unlinks Node p (a match) with predecessor trail.
	 */
	private void unlink(final Node p, final Node predTrail)
			throws InterruptedException {
		final ReentrantLock putLock = this.putLock;
		putLock.lockInterruptibly();
		try {
			if (p != predTrail) {
				predTrail.next = p.next;
			}
			p.item = null;
			if (last == p) {
				last = predTrail;
			}
		} finally {
			putLock.unlock();
		}
	}

	public Pair<Mailbox.MessageWrapper, Pattern> takeFirstMatch(final Pattern... predicates)
			throws InterruptedException {
		resetIfPatternsChanged(predicates);
		Pair<Mailbox.MessageWrapper, Pattern> ret = null;
		int c = -1;
		final AtomicInteger availableCount = this.numAvailable;
		final ReentrantLock takeLock = this.takeLock;
		while (ret == null) {
			takeLock.lockInterruptibly();
			try {
				/*
				 * Note that availableCount is used in this wait guard even
				 * though it is not protected by takeLock. This works because
				 * availableCount can only increase at this point (all other
				 * takes are shut out by takeLock), and we (or some other
				 * waiting take) are signaled if it ever gets above zero.
				 * Similarly for all other uses of availableCount in other wait
				 * guards.
				 */
				while (availableCount.get() == 0) {
					available.await();
				}
				ret = examineQueueForMatchFromReadPosition(predicates);
				if (ret != null) {
					c = availableCount.get();
					if (c > 0) {
						available.signal();
					}
				}
			} finally {
				takeLock.unlock();
			}				
		}
		return ret;
	}

	public Pair<Mailbox.MessageWrapper, Pattern> takeFirstMatchWithTimeout(
			long timeout, TimeUnit unit, final Pattern... predicates)
			throws TimeoutException, InterruptedException {
		resetIfPatternsChanged(predicates);
		Pair<Mailbox.MessageWrapper, Pattern> ret = null;
		long nanos = unit.toNanos(timeout);
		final AtomicInteger availableCount = this.numAvailable;
		final ReentrantLock takeLock = this.takeLock;
		while (ret == null) {
			takeLock.lockInterruptibly();
			try {
				if (nanos != 0L) {
					while (availableCount.get() == 0) {
						if (nanos < 0) {
							throw new TimeoutException();
						}
						nanos = available.awaitNanos(nanos);
					}
				}
				ret = examineQueueForMatchFromReadPosition(predicates);
				if (ret != null && availableCount.get() > 0) {
					available.signal();
				}
				if (ret == null) {
					if (nanos == 0L) {
						// If there is no matching message in the mailbox, the timeout will occur immediately.
						throw new TimeoutException();
					} else if (nanos > 0) {
						nanos = available.awaitNanos(nanos);
						// this is necessary for predicates that never match
						if (nanos <= 0) {
							throw new TimeoutException();
						}
					} else {
						throw new TimeoutException();
					}
				}
			} finally {
				takeLock.unlock();
			}				
		}
		return ret;
	}

	// callers must protect this by already holding the takeLock
	private Pair<Mailbox.MessageWrapper, Pattern> examineQueueForMatchFromReadPosition(final Pattern... predicates)
			throws InterruptedException {
		Node currentLast = null;
		Node targetNode = null;
		Pattern match = null;
		Pair<Mailbox.MessageWrapper, Pattern> ret = null;
		final ReentrantLock putLock = this.putLock;
		putLock.lockInterruptibly();
		try {
			currentLast = last;
		} finally {
			putLock.unlock();
		}
		if (currentLast == null) {
			throw new AssertionError();
		}

		Node predTrail = curr;
		for (Node trail = curr, p = trail.next; p != currentLast && p != null; trail = p, p = p.next) {
			predTrail = trail;
			final Mailbox.MessageWrapper item = p.item;
			if (item != null) {
				final Pattern pred = checkItemForMatch(item,
						predicates);
				if (pred != null) {
					targetNode = p;
					match = pred;
					break;
				}
			}
		}
		if (targetNode == null) {
			// no match found yet, examine currentLast for a possible match
			final Mailbox.MessageWrapper item = currentLast.item;
			if (item != null) {
				final Pattern pred = checkItemForMatch(item,
						predicates);
				if (pred != null) {
					if (currentLast != predTrail.next) {
						predTrail = currentLast;
					}
					targetNode = currentLast;
					match = pred;
				}
			}
		}
		if (targetNode != null) {
			// Found a matching node, so reset curr pointer to head.
			ret = Pair.create(targetNode.item, match);
			unlink(targetNode, predTrail);
			curr = head;
			numAvailable.decrementAndGet();
		} else if (curr != currentLast) {
			// Found no matching node and curr pointed to the left of
			// currentLast, thus set curr to currentLast
			curr = currentLast;
		}

		return ret;
	}

	private Pattern checkItemForMatch(final Mailbox.MessageWrapper item,
			final Pattern... predicates) {
		for (final Pattern predicate : predicates) {
			if (predicate instanceof SenderPidPattern) {
				if (!item.sender.equals(((SenderPidPattern) predicate).senderPid())) {
					continue;
				}
			}
			if (predicate.approve(item.payload)) {
				return predicate;
			}
		}
		return null;
	}
}
